#![allow(unused_variables)]

use std::collections::HashMap ;
use anyhow::{Result, anyhow} ;
use serde_json::{self, json, Value} ;
use async_trait::async_trait ;
use calamine::{open_workbook, Xlsx, Reader} ;
use itertools::Itertools;

pub mod types {
    use serde::{Serialize, Deserialize} ;
    use std::collections::HashMap ;
    use serde_json::Value ;

    #[derive(Debug, Default, Clone, Serialize, Deserialize)]
    pub struct DbCategory {
        pub tables: HashMap<DbTableId, DbTablePayload>,
        pub extra: Value, 
    }

    #[derive(Debug, Default, Clone, Serialize, Deserialize, Hash, Eq, PartialEq)]
    pub struct DbTableId (pub String) ;    

    #[derive(Debug, Default, Clone, Serialize, Deserialize)]
    pub struct DbTablePayload {
        pub name: String,
        pub display: Option<String>,
        pub columns: Vec<DbColPayload>,
        pub extra: Value, 
    }

    #[derive(Debug, Default, Clone, Serialize, Deserialize, Hash, Eq, PartialEq)]
    pub struct DbColId (pub String) ;

    #[derive(Debug, Default, Clone, Serialize, Deserialize)]
    pub struct DbColPayload {
        pub name: String,
        pub display: Option<String>,
        pub r#type: String,
        pub native_type: String,
        pub extra: Value,
    }
}

use types::* ;
use sled::Db ;
// use crate::common::types::Db ;

use crate::minio::types::* ;
use crate::minio::{self, MinioClient} ;

inventory::submit! {
    minio::MinioPlugin {
        new_client: | _db: Db, paths: Vec<String>| Box::pin(async {
            log::debug!("checking dataphine minio plugin...") ;
            let path_o = paths.into_iter().find(|x| x.starts_with("edw.")) ;
            if path_o.is_none() { return None } ;
            let path = path_o.unwrap() ;
            let id = path.split("/").collect_vec()[0] ;

            let client = EdwClient {client_id: id.to_string() } ;
            let box_client = Box::new(client) as Box<dyn MinioClient> ;
            Some(Ok(box_client))
        })
    }
}

impl DbCategory {

    pub async fn from_file(path: &str) -> Result<Self> {
        if path.ends_with(".xlsx") {
            let mut workbook: Xlsx<_> = open_workbook(path)? ;

            let table_dic =  workbook.worksheets()[0].1.rows().skip(1).map(|row|{
                let table_name = row[1].get_string().unwrap_or("").to_string() ;
                let table_comment = row[2].get_string().unwrap_or("").to_string() ;
                let col_name = row[3].get_string().unwrap_or("").to_string() ;
                let col_comment = row[4].get_string().unwrap_or("").to_string() ;
                let col_type = row[5].get_string().unwrap_or("") ;
                let hive_type = col_type.replace("unsigned", "").replace("zerofill", "").trim().to_uppercase() ;
                let hive_type = hive_type.split('(').collect_vec()[0] ;
                let hive_type = match hive_type {
                    "NUMBER" => "DECIMAL",
                    "LONG" => "BIGINT",
                    "DATE" => "DATETIME",
                    "TIME" => "TIMESTAMP",
                    "RAW" => "BINARY",
                    x if x.contains("LOB") => "STRING",
                    x if x.contains("CHAR") => "STRING",
                    x if x.contains("TEXT") => "STRING",
                    _ => hive_type,
                }.to_string() ;
                (table_name, table_comment, col_name, col_comment, col_type, hive_type)
            }).group_by(|x| x.0.clone()).into_iter().map(|(table_name, table_data)| {
                let mut table_data_peekable = table_data.peekable() ;
                let (_, table_comment, ..) = table_data_peekable.peek().unwrap().clone() ;
                let columns = table_data_peekable.map(|(_, _, col_name, col_comment, col_type, hive_type, ..)| {
                    DbColPayload {
                        name: col_name,
                        display: Some(col_comment),
                        native_type: col_type.to_string(),
                        extra: Value::Null,
                        r#type: hive_type,
                    }
                }).collect_vec() ;
                let table_id = DbTableId(table_name.to_uppercase()) ;
                let table_payload = DbTablePayload {
                    name: table_name,
                    display: Some(table_comment),
                    columns,
                    extra: Value::Null,
                } ;
                (table_id, table_payload)
            }).collect::<HashMap<_, _>>() ;
            let category = DbCategory {
                tables: table_dic,
                extra: Value::Null,
            } ;
            Ok(category)
        } else {
            unimplemented!()            
        }
    }
}
pub struct EdwClient {
    pub client_id: String
}
impl EdwClient {
    pub fn ddl(tpl_id: String, id: String) -> Result<String> {
        unimplemented!()
    }
    pub fn dml(tpl_id: String, id: String) -> Result<String> {
        unimplemented!()
    }
}

#[async_trait]
impl MinioClient for EdwClient {
    async fn mb(&self, req: MbReq) -> anyhow::Result<Value> {
        let MbReq { bucket, ..} = req ;
        let project_dir = &format!("./workspace/{}", bucket) ;
        std::fs::create_dir_all(&format!("{}/CATEGORY", project_dir))? ;
        std::fs::create_dir_all(&format!("{}/CONNECT", project_dir))? ;
        std::fs::create_dir_all(&format!("{}/CODE", project_dir))? ;
        std::fs::create_dir_all(&format!("{}/DAG", project_dir))? ;
        Ok(json!("SUCCESS"))
    }
    async fn ls(&self, req: LsReq) -> anyhow::Result<Value> {
        log::debug!("edw-ls: {:?}", req) ;
        let LsReq { path, ..} = req ;
        let path_parts_vec = path.split('/').collect_vec() ;
        if path_parts_vec.get(1).is_none() || path_parts_vec.get(1) != Some(&"CATEGORY") {
            return Err(anyhow!("only category ls supported!"))
        }
        let category_dir = format!("./workspace/{}/CATEGORY", self.client_id) ;
        let mut category_path_vec = std::fs::read_dir(category_dir)?.collect_vec() ;
        category_path_vec.sort_by_key(|it| {
            it.as_ref().unwrap().metadata().unwrap().modified().unwrap().clone()
        }) ;
        let category_path_vec = if path_parts_vec.get(2).is_none() || path_parts_vec.get(2) == Some(&"") {
            category_path_vec
        } else {
            let category_name = path_parts_vec.get(2).unwrap() ;
            let category_file = category_path_vec.into_iter().find(|it| it.as_ref().unwrap().file_name().into_string() == Ok(category_name.to_string())) ;
            if category_file.is_none() {
                return Err(anyhow!("category not find!"))
            }
            vec![category_file.unwrap()]
        };
        return Ok(json!(category_path_vec.into_iter().flat_map(|entry| {
            let entry = entry.unwrap() ;
            let category = serde_json::from_reader::<_, DbCategory>(std::fs::File::open(entry.path()).unwrap()).unwrap() ;
            let mut tables = category.tables.into_keys().into_iter().map(|id| {
                format!("{}@{}", id.0, entry.file_name().into_string().unwrap())
            }).collect_vec() ;
            tables.sort() ;
            tables
        }).collect_vec()))
    }
    async fn cp(&self, req: CpReq) -> anyhow::Result<Value> {
        let source_path = &req.source;
        let target_path = &req.sink ;
        if source_path.starts_with("./") || source_path.starts_with("/") {
            let db_category = &DbCategory::from_file(source_path).await? ;
            let tee_req = TeeReq { data: serde_json::to_vec_pretty(db_category)?, path: target_path.to_owned() } ;
            return self.tee(tee_req).await
        }
        Ok(json!("unimplemented!"))
    }
    async fn cat(&self, req: CatReq) -> anyhow::Result<Value> {
        let CatReq {path, ..} = req ;

        let path_parts_vec = path.split('/').collect_vec() ;
        let schema_name = path_parts_vec.get(0).unwrap() ;        
        if path_parts_vec.get(1).is_none() || path_parts_vec.get(1) != Some(&"VIEW") {
            return Err(anyhow!("only view cat supported!"))
        }
        if path_parts_vec.get(2).is_none() || path_parts_vec.get(2) == Some(&"") {
            return Err(anyhow!("[tpl_id].[dw_id] needed!"))
        }
        let tpl_code = path_parts_vec.get(2).unwrap() ;
        let (tpl_id, dw_id) = tpl_code.splitn(2, '.').collect_tuple().unwrap() ;
        let category_dir = format!("./workspace/{}/CATEGORY", schema_name) ;
        let mut category_path_vec = std::fs::read_dir(category_dir)?.collect_vec() ;
        category_path_vec.sort_by_key(|it| {
            it.as_ref().unwrap().metadata().unwrap().modified().unwrap().clone()
        }) ;
        let table_jsons = category_path_vec.into_iter().filter_map(|it| {
            let category = serde_json::from_reader::<_, DbCategory>(std::fs::File::open(it.unwrap().path()).unwrap()).unwrap() ;
            let table_id = &DbTableId(dw_id.to_uppercase()) ;
            category.tables.get(table_id).map(|it| it.to_owned())
        }).collect_vec() ;
        if table_jsons.is_empty() {
            return Err(anyhow!(format!("{} NOT FOUND", tpl_code)))
        }
        let table_json = &table_jsons[0] ;
        log::debug!("tpl_id: {}", tpl_id) ;
        if tpl_id == "json" {
            return Ok(serde_json::to_value(table_json).unwrap()) ;
        }
        let tpl_path = format!("workspace/_tpl_/{}.{}.sql.tpl", self.client_id, tpl_id) ;
        let tpl_text = std::fs::read_to_string(tpl_path)?  ;
        let mut reg = handlebars::Handlebars::new();
        reg.register_template_string(tpl_id, tpl_text)? ;
        let sql = reg.render(tpl_id, &serde_json::to_value(&table_json).unwrap())? ;
        Ok(json!(sql))
    }
    async fn tee(&self, req: TeeReq) -> anyhow::Result<Value> {
        let TeeReq {path, data, ..}  = req ;

        let path_parts_vec = path.split('/').collect_vec() ;
        if path_parts_vec.get(1).is_none() || path_parts_vec.get(1) != Some(&"CATEGORY") {
            return Err(anyhow!("only category tee supported!"))
        }
        if path_parts_vec.get(2).is_none() || path_parts_vec.get(2) == Some(&"") {
            return Err(anyhow!("file name needed!"))
        }
        let file_name = path_parts_vec.get(2).unwrap() ;
        let path = format!("./workspace/{}/CATEGORY/{}", self.client_id, file_name) ;
        std::fs::write(path, data)? ;
        Ok(json!("SUCCESS"))
    }
    async fn rm(&self, req: RmReq) -> anyhow::Result<Value> {
        Ok(json!("unimplemented!"))
    }

    async fn share(&self, req: ShareReq) -> anyhow::Result<Value> {
        Ok(json!("unimplemented!"))
    }
    async fn mirror(&self, req: MirrorReq) -> anyhow::Result<Value> {
        unimplemented!()
    }
    async fn sql(&self, req: SqlReq) -> anyhow::Result<Value> {
        unimplemented!()
    }
}
